package streamAPI.asyncio;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.util.JdbcUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import util.FlinkConstant._MySQL;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * @author yue.cao
 * @since 10-21-2020
 */
@Slf4j
public class AsyncMySQLRequet extends RichAsyncFunction<String, Tuple2<String, String>> {

	private transient ExecutorService executorService;
	private transient DruidDataSource dataSource;

	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		executorService = Executors.newFixedThreadPool(10);
		dataSource = new DruidDataSource();
		dataSource.setDriverClassName("com.mysql.jdbc.Driver");
		dataSource.setUsername(_MySQL.USER_NAME);
		dataSource.setPassword(_MySQL.PASSWORD);
		dataSource.setUrl(_MySQL.URL);
		dataSource.setInitialSize(1);
		dataSource.setMaxActive(2);
	}

	@Override
	public void asyncInvoke(String input, ResultFuture<Tuple2<String, String>> resultFuture) throws Exception {
		Future<String> future = executorService.submit(new Callable<String>() {
			@Override
			public String call() throws Exception {
				Connection connection = null;
				PreparedStatement statement = null;
				ResultSet rs = null;
				String res = null;
				try {
					String sql = "SELECT id FROM t_activities WHERE a_name = ?";
					connection = dataSource.getConnection();
					statement = connection.prepareStatement(sql);
					statement.setString(1, "一刀999");
					rs = statement.executeQuery();
					while (rs.next()) {
						res = rs.getString("id");
					}
					return res;
				} catch (Exception e) {
					return null;
				} finally {
					JdbcUtils.close(rs);
					JdbcUtils.close(statement);
					JdbcUtils.close(connection);
				}
			}
		});

		CompletableFuture.supplyAsync(new Supplier<String>() {
			@Override
			public String get() {
				try {
					return future.get();
				} catch (Exception e) {
					log.error("error", e);
					return null;
				}
			}
		}).thenAccept(new Consumer<String>() {
			@Override
			public void accept(String res) {
				resultFuture.complete(Collections.singleton(Tuple2.of(input, res)));
			}
		});
	}

	@Override
	public void close() throws Exception {
		super.close();
		dataSource.close();
		executorService.shutdown();
	}
}
